package com.yanduo.utils

import org.apache.commons.lang3.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.Row

import scala.collection.mutable
import scala.collection.mutable.ListBuffer

/**
  *
  * @author Gerry chan
  *  2020/5/5 11:04
  * @version 1.0
  */
object TagsUtils {
  val hasSomeUserIdCondition=  """
                                 |imei !="" or imeimd5!="" or imeisha1!=""
                                 |idfa !="" or idfamd5 !="" or idfasha1!=""
                                 |adroidid !="" or adnroididmd5!="" or androididsha1 !=""
                                 |openudid !="" or openudidmd5!="" or openudidsha1 !=""
                                 |
      """.stripMargin

  /**
    * 获取用户某一个不为空的id
    * @param v
    * @return
    */
  def getAllUserId(v:Row):ListBuffer[String] = {
    val userIds = new collection.mutable.ListBuffer[String]()

    if (StringUtils.isNotEmpty(v.getAs[String]("imei")))  userIds.append("IM#"+v.getAs[String]("imei"))
    if (StringUtils.isNotEmpty(v.getAs[String]("imei")))  userIds.append("IM#"+v.getAs[String]("imei"))
    if (StringUtils.isNotEmpty(v.getAs[String]("idfa")))  userIds.append("ID#"+v.getAs[String]("idfa"))
    if (StringUtils.isNotEmpty(v.getAs[String]("mac")))  userIds.append("MC#"+v.getAs[String]("mac"))
    if (StringUtils.isNotEmpty(v.getAs[String]("androidid")))  userIds.append("AD#"+v.getAs[String]("androidid"))
    if (StringUtils.isNotEmpty(v.getAs[String]("openudid")))  userIds.append("OP#"+v.getAs[String]("openudid"))
    if (StringUtils.isNotEmpty(v.getAs[String]("imeimd5")))  userIds.append("IMM#"+v.getAs[String]("imeimd5"))
    if (StringUtils.isNotEmpty(v.getAs[String]("idfamd5")))  userIds.append("IDM#"+v.getAs[String]("idfamd5"))
    if (StringUtils.isNotEmpty(v.getAs[String]("macmd5")))  userIds.append("MCM#"+v.getAs[String]("macmd5"))
    if (StringUtils.isNotEmpty(v.getAs[String]("androididmd5")))  userIds.append("ADM#"+v.getAs[String]("androididmd5"))
    if (StringUtils.isNotEmpty(v.getAs[String]("openudidmd5")))  userIds.append("OPM#"+v.getAs[String]("openudidmd5"))
    if (StringUtils.isNotEmpty(v.getAs[String]("imeisha1")))  userIds.append("IMS#"+v.getAs[String]("imeisha1"))
    if (StringUtils.isNotEmpty(v.getAs[String]("idfasha1")))  userIds.append("IDS#"+v.getAs[String]("idfasha1"))
    if (StringUtils.isNotEmpty(v.getAs[String]("macsha1")))  userIds.append("MCS#"+v.getAs[String]("macsha1"))
    if (StringUtils.isNotEmpty(v.getAs[String]("androididsha1")))  userIds.append("ADS#"+v.getAs[String]("androididsha1"))
    if (StringUtils.isNotEmpty(v.getAs[String]("openudidsha1")))  userIds.append("OPS#"+v.getAs[String]("openudidsha1"))

    userIds
  }
  /**
    * 获取当前行的用户便签数据，返回当前行的一组标签数据
    * @param row
    * @return
    */
  def getCurrentLineTags(row: Row,
                         appdict: Broadcast[Map[String, String]],
                         stopdict: Broadcast[Map[String, Null]]) : mutable.HashMap[String, Int]= {

    val map = new mutable.HashMap[String, Int]()

    val adSpt = row.getAs[Int]("adspacetype")
    if (adSpt>0 && adSpt < 10) map.put("LC0" + adSpt, 1) else map.put("LC" + adSpt, 1)
    val adSptn = row.getAs[String]("adspacetypename")
    if (StringUtils.isNotEmpty(adSptn)) map.put("LN" + adSptn, 1)


    val appId = row.getAs[String]("appid")
    val appName = row.getAs[String]("appname")

    if (StringUtils.isEmpty(appName)) {
      if (StringUtils.isNotEmpty(appId)) {
        // 从广播变量中获取
        val newAppName = appdict.value.getOrElse(appId, appId)
        map.put("APP"+newAppName, 1)
      }
    } else map.put("APP"+appName, 1)

    // 渠道
    val channelId = row.getAs[Int]("adplatformproviderid")
    if (channelId > 0) map.put("CN"+channelId, 1)

    // 设备
    val client = row.getAs[Int]("client")
    //获取设备联网方式字段
    val net = row.getAs[String]("networkmannername").toUpperCase
    //获取运营商名称字段
    val opera = row.getAs[String]("ispname")

    client match {
      case 1 => map.put("D00010001", 1)
      case 2 => map.put("D00010002", 1)
      case 3 => map.put("D00010003", 1)
      case _ => map.put("D00010004", 1)
    }

    net match {
      case "WIFI" => map.put("D00020001", 1)
      case "4G" => map.put("D00020002", 1)
      case "3G" => map.put("D00020003", 1)
      case "2G" => map.put("D00020004", 1)
      case _ => map.put("D00020005", 1)
    }

    opera match {
      case "移动" => map.put("D00030001", 1)
      case "联通" => map.put("D00030002", 1)
      case "电信" => map.put("D00030003", 1)
      case _ => map.put("D00030004", 1)
    }


    //获取关键字字段
    val keywords = row.getAs[String]("keywords")
    if (StringUtils.isNotEmpty(keywords)) {
      keywords.split("\\|")
        .filter(word => word.length >= 3 && word.length <= 8) // 长度的符合要求
        .filter(word => !stopdict.value.contains(word)) // 判断是否是否在停用词库中，如果存在则不要
        .foreach(word => map.put("K"+word, 1))
    }


    // 地域标签
    val pName = row.getAs[String]("provincename")
    val cName = row.getAs[String]("cityname")

    if (StringUtils.isNotEmpty(pName)) map.put("ZP"+pName, 1)
    if (StringUtils.isNotEmpty(cName)) map.put("ZC"+cName, 1)

    map
  }
}
